package com.fxkj.cloud.drms.infra.config;

import com.fxkj.cloud.drms.infra.mq.mqtt.MqttConstants;
import com.fxkj.cloud.drms.infra.mq.mqtt.MqttExecutorService;
import com.fxkj.cloud.drms.mqtt.*;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;

import java.util.function.Consumer;

/**
 * <p>description: mqtt客户端初始化</p>
 * @author ww
 * @date 2025/3/14 10:37
 */
@Configuration
@Slf4j
public class MqttConfiguration {

    //mqtt参数配置类
    private final MqttProperties mqttProperties;
    //mqtt消息处理线程池
    private final MqttExecutorService mqttExecutorService;
    //主题处理消费者
    private final Consumer<MqttClient> mqttSubscriberConsumer;
    //初始化Netty的事件循环
    private final EventLoopGroup eventExecutors = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);

    public MqttConfiguration(MqttProperties mqttProperties,
                             @Lazy MqttHandler mqttHandler,
                             MqttExecutorService mqttExecutorService) {
        this.mqttProperties = mqttProperties;
        this.mqttExecutorService = mqttExecutorService;
        this.mqttSubscriberConsumer = mqttClient -> {
            for (String subscribeTopic : MqttConstants.Topic.SUBSCRIBE_TOPICS) {
                mqttClient.on(subscribeTopic, mqttHandler);
                log.info("mqtt 订阅主题: {}", subscribeTopic);
            }
        };
    }

    @PostConstruct
    public void init() {
        log.info("[fx-cloud] -| MQTT 客户端配置加载完成!");
    }

    @Bean
    public MqttClient mqttClient() {
        //初始化mqtt客户端配置
        MqttClientConfig mqttClientConfig = new MqttClientConfig();
        mqttClientConfig.setUsername(mqttProperties.getClient().getUsername());
        mqttClientConfig.setPassword(mqttProperties.getClient().getPassword());
        MqttClient mqttClient = MqttClient.create(
                mqttClientConfig,
                null,
                mqttExecutorService
        );
        //配置事件循环
        mqttClient.setEventLoop(eventExecutors);
        MqttClientCallback callback = new MqttClientCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                log.info("mqtt connect lost,cause:{}", cause.getMessage());
            }

            @Override
            public void onSuccessfulReconnect() {
                log.info("mqtt reconnect success");
                //重新订阅主题,连接断开过后订阅的主题会清空,所以进行重新订阅
                try {
                    int reTryTimes = 3;
                    while ((reTryTimes--) > 0) {
                        Thread.sleep(3000);
                        if (mqttClient.isConnected()) {
                            mqttSubscriberConsumer.accept(mqttClient);
                        }
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        };
        mqttClient.setCallback(callback);
        doConnect(mqttClient, mqttProperties.getHost(), mqttProperties.getClient().getPort(), mqttSubscriberConsumer);
        return mqttClient;
    }

    /**
     * 进行mqtt连接
     * @param client
     * @param host
     * @param port
     * @param consumer
     */
    private void doConnect(MqttClient client, String host, Integer port, Consumer<MqttClient> consumer) {
        Future<MqttConnectResult> connect = client.connect(host, port);
        //添加MQTT连接完成的监听器回调,待MQTT连接完成且成功时进行主题的订阅
        connect.addListener(connectResult -> {
            if (connectResult.isSuccess()) {
                log.info("[fx-cloud] -| mqtt客户端初始化完毕,连接成功! broker=> {}:{}", host, port);
                //订阅主题
                //1.更新完成主题
                consumer.accept(client);
            } else {
                log.warn("[fx-cloud] -| mqtt客户端初始化失败!broker=> {}:{}", host, port);
            }
        });
    }
}
